package org.jetlinks.pro.things.impl.rpc;

import lombok.AllArgsConstructor;
import org.jetlinks.core.message.ThingMessage;
import org.jetlinks.core.things.MetadataId;
import org.jetlinks.core.things.ThingId;
import org.jetlinks.core.things.ThingRpcSupport;
import org.jetlinks.pro.things.impl.connector.AbstractThingsConnector;
import org.jetlinks.pro.things.utils.ThingsUtils;
import reactor.core.Disposable;
import reactor.core.publisher.EmitterProcessor;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import reactor.core.publisher.Mono;

import java.util.function.Function;


/**
 *  物RPC连接器，用于处理RPC调用
 *
 * @author zhouhao
 * @since 1.12
 */
@AllArgsConstructor
public class ThingsRpcConnector extends AbstractThingsConnector implements ThingRpcSupport {

    //RPC支持
    private final ThingRpcSupport support;

    //物消息处理器，用于接收到物消息后流向下游处理
    private final EmitterProcessor<ThingMessage> processor = EmitterProcessor.create(false);

    //支持从多个线程中拿到物消息放到物消息处理器中
    private final FluxSink<ThingMessage> sink = processor.sink(FluxSink.OverflowStrategy.BUFFER);


    /**
     * 连接到指定物的物模型,当产生消息时,handler将被调用.可调用返回值的{@link Disposable#dispose()}来断开连接
     *
     * @param thingId    物ID
     * @param metadataId 物模型ID
     * @param handler    消息处理器
     * @return Disposable
     */
    @Override
    public Disposable doConnectTo(ThingId thingId,
                                MetadataId metadataId,
                                Function<ThingMessage, Mono<Void>> handler) {
        return processor
            .flatMap(handler)
            .subscribe();
    }

    /**
     * 连接到当前物的物模型，物实例ID由数据源生成
     * 当产生消息时,handler将被调用.可调用返回值的{@link Disposable#dispose()}来断开连接
     *
     * @param metadataId 物模型ID
     * @param handler    消息处理器
     * @return Disposable
     */
    @Override
    public Disposable doConnectTo(MetadataId metadataId,
                                Function<ThingMessage, Mono<Void>> handler) {
        return processor
            .flatMap(handler)
            .subscribe();
    }
    /**
     * RPC调用
     *
     * @param message RPC请求消息
     * @return 多个RPC的调用结果
     */
    @Override
    public Flux<? extends ThingMessage> call(ThingMessage message) {
        //当处理器存在订阅者时，放入数据到处理器中
        if (processor.hasDownstreams()) {
            sink.next(message);
        }
        return support
            .call(message)
            .doOnNext(response -> handleResponse(message, response));
    }


    private void handleResponse(ThingMessage request, ThingMessage response) {
        //合并请求的header到响应中
        ThingsUtils.mergeHeader(request, response);
        //当处理器存在订阅者时，放入数据到处理器中
        if (processor.hasDownstreams()) {
            sink.next(response);
        }
    }

    @Override
    protected void onDispose() {
        if (support instanceof Disposable) {
            ((Disposable) support).dispose();
        }
        processor.onComplete();
        sink.complete();
    }

}
